package graph

import java.sql.DriverManager

import Configer.Config
import org.apache.spark.graphx.{Edge, Graph}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}
import tag._
import util.GetUUID

import scala.collection.mutable.ListBuffer

//标签合并图计算
object GraphTags {
  def main(args: Array[String]): Unit = {
    //sparkContext
    val conf = new SparkConf()
    conf.setAppName(s"${this.getClass.getName}")
    conf.setMaster("local[*]")
    conf.set("spark.serializer", Config.serializer)

    val sc = new SparkContext(conf)

    val sQLContext = new SQLContext(sc)

    //读取数据
    val dataFrame = sQLContext.read.parquet(Config.parquetPath)
    val dictLines = sc.textFile("C:\\Users\\44323\\Desktop\\资料PDF\\app_dict.txt")
    val stopLines = sc.textFile("C:\\Users\\44323\\Desktop\\资料PDF\\stopwords.txt")


    //广播变量
    val dictMap = dictLines.map(_.split("\t", -1)).filter(_.length >= 5).map(arr => (arr(4), arr(1))).collectAsMap()
    val dictBc = sc.broadcast(dictMap)

    val stopMap = stopLines.map((_, null)).collectAsMap()
    val stopBc = sc.broadcast(stopMap)

    //点集合
    val point: RDD[(Long, List[(String, Int)])] = dataFrame.filter(
      """
imei != "" or mac != "" or idfa != "" or openudid != "" or androidid != "" or
imeimd5 != "" or macmd5 != "" or idfamd5 != "" or openudidmd5 != "" or androididmd5 != "" or
imeisha1 != "" or macsha1 != "" or idfasha1 != "" or openudidsha1 != "" or androididsha1 != ""
      """.stripMargin).flatMap(row => {
      //获取所有Id
      val allUserIDs = GetUUID.getUUID(row)
      //广告位
      val listAd: List[(String, Int)] = AdTags.makeTags(row)

      //app
      val listApp: List[(String, Int)] = AppTags.makeTags(row, dictBc)

      //渠道
      val listChan = ChannelTags.makeTags(row)

      //设备  操作系统  联网方式  运营商
      val listDev = DeviceTags.makeTags(row)
      //关键字
      val listKey = KeyTags.makeTags(row, stopBc)

      //地域标签
      val listArea = AreaTags.makeTags(row)

      //商圈
      val conn = DriverManager.getConnection(Config.url, Config.user, Config.password)
      val listBusiness = BusinessTags.makeTags(row, conn)
      conn.close()

      val ids: ListBuffer[(String, Int)] = allUserIDs.map((_, 1))
      val list = listAd ++ listApp ++ listChan ++ listDev ++ listKey ++ listArea ++ listBusiness ++ ids
      allUserIDs.map(id => {
        if (id.equals(allUserIDs.head)) {
          (id.hashCode.toLong, list)
        } else {
          (id.hashCode.toLong, List.empty)
        }
      })
    })
    point


    //线集合
    val edge: RDD[Edge[Int]] = dataFrame.filter(
      """
imei != "" or mac != "" or idfa != "" or openudid != "" or androidid != "" or
imeimd5 != "" or macmd5 != "" or idfamd5 != "" or openudidmd5 != "" or androididmd5 != "" or
imeisha1 != "" or macsha1 != "" or idfasha1 != "" or openudidsha1 != "" or androididsha1 != ""
      """.stripMargin).flatMap(row => {
      //获取所有Id
      val allUserIDs = GetUUID.getUUID(row)
      allUserIDs.map(id => Edge(allUserIDs.head.hashCode.toLong, id.hashCode.toLong, 0))
    })
    edge

    //图对象
    val graph = Graph(point,edge)
    val ver = graph.connectedComponents().vertices


    //聚合
    ver.join(point).map{
      case (id,(mid,info)) => (mid,info)
    }.reduceByKey{
      (list1,list2)=>(list1++list2).groupBy(_._1)//.mapValues(_.map(_._2).sum).toList
          .mapValues(_.foldLeft(0)(_+_._2)).toList
    }.coalesce(1).saveAsTextFile("C:\\Users\\44323\\Desktop\\资料PDF\\GraphTag")

    //释放资源
    sc.stop()
  }
}
